{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " # Feature Store Tour - Scala API\n",
    " \n",
    "This notebook contains a tour/reference for the feature store Scala API on hopsworks. We will go over best practices for using the API as well as common pitfalls.\n",
    " \n",
    "The notebook is designed to be used in combination with the Feature Store Tour on Hopsworks, it assumes that you have run the following feature engineering job: [job](https://github.com/logicalclocks/hops-examples/tree/master/featurestore) (**the job is added automatically when you start the feature store tour in Hopsworks. You can run the job by going to the 'Jobs' tab to the left in the Hopsworks project home page**). \n",
    "\n",
    "Which will produce the following model of feature groups in your project's feature store:\n",
    "\n",
    "![Feature Store Model](./images/model.png \"Feature Store Model\")\n",
    "\n",
    "In this notebook we will run queries over this feature store model. We will also create new feature groups and training datasets.\n",
    "\n",
    "We will go from (1) features to (2) training datasets to (3) A trained model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Store 101\n",
    "\n",
    "The simplest way to think about the feature store is as a central place to store curated /features/ within an organization. A feature is a measurable property of some phenomenon. It could be for example an image-pixel, a word from a piece of text, the age of a person, a coordinate emitted from a sensor, or an aggregate value like the average number of purchases within the last hour.\n",
    "\n",
    "A feature store is a data management layer for machine learning that can optimize the machine learning workflow and provide an interface between data engineering and data science.\n",
    "\n",
    "![Feature Store Overview](./images/overview.png \"Feature Store Overview\")\n",
    "\n",
    "A feature store is not a pure service concept, it goes hand-in-hand with feature computation. Feature engineering is the process of transforming raw data into a format that is compatible and understandable for predictive models.\n",
    "\n",
    "There are two interfaces to the feature store:\n",
    "\n",
    "- Writing to the feature store, at the end of the feature engineering pipeline the features are written to the feature store, e.g:\n",
    "\n",
    "```scala\n",
    "val rawData = spark.read.format(\"csv\").load(filename)\n",
    "val polynomialFeatures = rawData.map((x: Float) => scala.math.pow(x, 2))\n",
    "import io.hops.util.Hops\n",
    "Hops.insertIntoFeaturegroup(\"polynomial_features\").setDataframe(polynomialFeatures).write()\n",
    "```\n",
    "- Reading from the feature store, to train a model on a set of features, the features can be read from the feature store, e.g:\n",
    "\n",
    "```scala\n",
    "import io.hops.util.Hops\n",
    "val features = List(\"team_budget\", \"average_attendance\", \"average_player_age\")\n",
    "val featuresDf = Hops.getFeatures(spark, features, Hops.getProjectFeaturestore)\n",
    "```\n",
    "\n",
    "As a data engineer/data scientist, you can think of the feature store as a middle-layer. Once you have computed a set of features, instead of writing them locally to a csv file, insert them in the feature store so that the features can get documented/versioned, backfilled, **and so that your colleagues can re-use your features!** \n",
    "\n",
    "In this notebook we will take a look at interacting with the Feature Store through the Scala SDK,there is also a Python SDK available if you prefer that."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The hops library is automatically installed in all Hopsworks-projects.You can find API documentation [here](http://snurran.sics.se/hops/hops-util-javadoc/0.6.0-SNAPSHOT/)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>48</td><td>application_1573462172517_0016</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-22-2.eu-north-1.compute.internal:8088/proxy/application_1573462172517_0016/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-22-2.eu-north-1.compute.internal:8042/node/containerlogs/container_e03_1573462172517_0016_01_000001/demo_featurestore_admin000__meb10000\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n",
      "import io.hops.util.Hops\n",
      "import scala.collection.JavaConversions._\n",
      "import collection.JavaConverters._\n"
     ]
    }
   ],
   "source": [
    "import io.hops.util.Hops\n",
    "import scala.collection.JavaConversions._\n",
    "import collection.JavaConverters._"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How The Scala API Works\n",
    "\n",
    "All operations are lazy, meaning that when you call `operation1()` you get back a DTO object where you can set the different parameters of your operation (or let them be the default values). To execute the operation call `.read()` or `.write()` on the DTO."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get The Name of The Project's Feature Store\n",
    "\n",
    "Each project with the feature store service enabled automatically gets its own feature store created. This feature store is only accessible within the project unless you decide to share it with other projects. The name of the feature store is `<project_name>_featurestore`, and you can get the name with the API method `getProjectFeaturestore`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res1: String = demo_featurestore_admin000_featurestore\n"
     ]
    }
   ],
   "source": [
    "Hops.getProjectFeaturestore().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get a List of All Feature Stores Accessible in the Current Project \n",
    "\n",
    "Feature Stores can be shared across projects in a multi-tenant manner, just like any Hopsworks-dataset can. You can read more about sharing datasets at [hops.io](hops.io), but in essence to share a dataset you just have to right click on it in your project. The features and featuregroups in the feature store are stored in a dataset called `<project_name>_featurestore.db` in your project.\n",
    "\n",
    "![Share Feature Store](./images/share_featurestore.png \"Share Feature Store\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res2: java.util.List[String] = [demo_featurestore_admin000_featurestore]\n"
     ]
    }
   ],
   "source": [
    "Hops.getProjectFeaturestores().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Querying The Feature Store\n",
    "\n",
    "The feature store can be queried programmatically and with raw SQL. When you query the feature store programmatically, the library will infer how to fetch the different features using a **query planner**. \n",
    "\n",
    "![Feature Store Query Planner](./images/query_optimizer.png \"Feature Store Query Planner\")\n",
    "\n",
    "When interacting with the feature store it is sufficient to be familiar with three concepts:\n",
    "\n",
    "- The **feature**, this refer to an individual versioned and documented feature in the feature store, e.g the age of a person.\n",
    "- The **feature group**, this refer to a documented and versioned group of features stored as a Hive table that is linked to a specific Spark/Numpy/Pandas job that takes in raw data and outputs the computed features.\n",
    "- The **training dataset**, this refer to a versioned and managed dataset of features, stored in HopsFS as tfrecords, .csv, .tsv, or parquet.\n",
    "\n",
    "A feature group contains a group of features and a training dataset contains a set of features, potentially from many different feature groups.\n",
    "\n",
    "![Feature Store Concepts](./images/concepts.png \"Feature Store Contents\")\n",
    "\n",
    "When you query the feature store you will always get back the results in a spark dataframe. This is for scalability reasons. If the dataset is small and you want to work with it in memory you can convert it into a pandas dataframe or a numpy matrix using one line of code as we will demonstrate later on in this notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fetch an Individual Feature\n",
    "\n",
    "When retrieving a single feature from the featurestore, the hops-util-py library will infer in which feature group the feature belongs to by querying the metastore, but you can also explicitly specify which featuregroup and version to query. \n",
    "\n",
    "If there are multiple features of the same name in the featurestore, it is required to specify enough information to uniquely identify the feature (e.g specify feature group and version). If no featurestore is provided it will default to the project's featurestore.\n",
    "\n",
    "To read an individual feature, use the method `getFeature`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|team_budget|\n",
      "+-----------+\n",
      "|  12957.076|\n",
      "|  2403.3704|\n",
      "|  3390.3755|\n",
      "|  13547.429|\n",
      "|   9678.333|\n",
      "+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeature(\"team_budget\").read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also explicitly specify the arguments: spark-session, feature store, feature group and version (getFeature returns a builder object that you can populate with parameters using *setParameterName*):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|team_budget|\n",
      "+-----------+\n",
      "|  12957.076|\n",
      "|  2403.3704|\n",
      "|  3390.3755|\n",
      "|  13547.429|\n",
      "|   9678.333|\n",
      "+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeature(\"team_budget\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setFeaturegroup(\"teams_features\").setVersion(1).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fetch an Entire Feature Group\n",
    "\n",
    "You can get an entire featuregroup from the API. If no feature store is provided the API will default to the project's feature store, if no version is provided it will default to version 1 of the feature group."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+-------------+\n",
      "|team_budget|team_id|team_position|\n",
      "+-----------+-------+-------------+\n",
      "|  12957.076|      1|            1|\n",
      "|  2403.3704|      2|            2|\n",
      "|  3390.3755|      3|            3|\n",
      "|  13547.429|      4|            4|\n",
      "|   9678.333|      5|            5|\n",
      "+-----------+-------+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"teams_features\").read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Explicitly specifying the arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+-------------+\n",
      "|team_budget|team_id|team_position|\n",
      "+-----------+-------+-------------+\n",
      "|  12957.076|      1|            1|\n",
      "|  2403.3704|      2|            2|\n",
      "|  3390.3755|      3|            3|\n",
      "|  13547.429|      4|            4|\n",
      "|   9678.333|      5|            5|\n",
      "+-----------+-------+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"teams_features\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setVersion(1).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fetch A Set of Features\n",
    "\n",
    "When retrieving a list of features from the featurestore, the hops-util-py library will infer which featuregroup the features belongs to by querying the metastore. If the features reside in different featuregroups, the library will also try to infer how to join the features together based on common columns. If the JOIN query cannot be inferred due to existence of multiple features with the same name or non-obvious JOIN query, the user need to supply enough information to the API call to be able to query the featurestore. If the user already knows the JOIN query it can also run featurestore.sql(joinQuery) directly (an example of this is shown further down in this notebook). If no featurestore is provided the API will default to the project's featurestore."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Example of querying the feature store for a list of features without specifying the feature groups and feature store:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------------+------------------+\n",
      "|team_budget|average_attendance|average_player_age|\n",
      "+-----------+------------------+------------------+\n",
      "|    7307.94|         19595.973|             24.85|\n",
      "|   7326.092|          6462.462|             25.45|\n",
      "|   3555.235|          7226.672|              25.4|\n",
      "|  910.39325|         3189.8455|             25.91|\n",
      "|  12474.419|          9405.213|             25.78|\n",
      "+-----------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeatures(List(\"team_budget\", \"average_attendance\", \"average_player_age\")).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also explicitly specify the feature groups where the features reside. Either the feature groups and versions can be specified by prepending feature names with `<feature group name>_<feature group version.`, or by providing a Map with entries of `<feature group name> -> <feature group version>`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "features: List[String] = List(teams_features_1.team_budget, attendances_features_1.average_attendance, players_features_1.average_player_age)\n",
      "+-----------+------------------+------------------+\n",
      "|team_budget|average_attendance|average_player_age|\n",
      "+-----------+------------------+------------------+\n",
      "|    7307.94|         19595.973|             24.85|\n",
      "|   7326.092|          6462.462|             25.45|\n",
      "|   3555.235|          7226.672|              25.4|\n",
      "|  910.39325|         3189.8455|             25.91|\n",
      "|  12474.419|          9405.213|             25.78|\n",
      "+-----------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "val features = List(\"teams_features_1.team_budget\", \n",
    "                    \"attendances_features_1.average_attendance\", \n",
    "                    \"players_features_1.average_player_age\")\n",
    "Hops.getFeatures(features).setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).read().show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "featuregroupsMap: scala.collection.immutable.Map[String,Integer] = Map(teams_features -> 1, attendances_features -> 1, players_features -> 1)\n",
      "javaFeaturegroupsMap: java.util.HashMap[String,Integer] = {attendances_features=1, players_features=1, teams_features=1}\n",
      "+-----------+------------------+------------------+\n",
      "|team_budget|average_attendance|average_player_age|\n",
      "+-----------+------------------+------------------+\n",
      "|    7307.94|         19595.973|             24.85|\n",
      "|   7326.092|          6462.462|             25.45|\n",
      "|   3555.235|          7226.672|              25.4|\n",
      "|  910.39325|         3189.8455|             25.91|\n",
      "|  10290.323|         4964.6475|             26.63|\n",
      "+-----------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "val featuregroupsMap = Map[String, Integer](\n",
    "    \"teams_features\"->1,\n",
    "    \"attendances_features\"->1,\n",
    "    \"players_features\"->1\n",
    ")\n",
    "val javaFeaturegroupsMap = new java.util.HashMap[String, Integer](featuregroupsMap)\n",
    "Hops.getFeatures(features).setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setFeaturegroupsAndVersions(javaFeaturegroupsMap).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you have a lot of name collisions and it is not obvious how to infer the JOIN query to get the features from the feature store. You can explicitly specify the argument `joinKey` to the API (or you can provide the entire SQL query using the API method `.sql` as we will demonstrate later on in the notebook)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------------+------------------+\n",
      "|team_budget|average_attendance|average_player_age|\n",
      "+-----------+------------------+------------------+\n",
      "|    7307.94|         19595.973|             24.85|\n",
      "|   7326.092|          6462.462|             25.45|\n",
      "|   3555.235|          7226.672|              25.4|\n",
      "|  910.39325|         3189.8455|             25.91|\n",
      "|  10290.323|         4964.6475|             26.63|\n",
      "+-----------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeatures(features).setJoinKey(\"team_id\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setFeaturegroupsAndVersions(javaFeaturegroupsMap).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Advanced Eamples of Fetching Sets of Features and Common Pitfalls"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Getting 12 features from 4 different feature groups:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "features1: List[String] = List(team_budget, average_attendance, average_player_age, team_position, sum_attendance, average_player_rating, average_player_worth, sum_player_age, sum_player_rating, sum_player_worth, sum_position, average_position)\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|team_budget|average_attendance|average_player_age|team_position|sum_attendance|average_player_rating|average_player_worth|sum_player_age|sum_player_rating|sum_player_worth|sum_position|average_position|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|    7307.94|         19595.973|             24.85|            6|     391919.47|            1311.2384|           1435.2465|        2485.0|        131123.84|       143524.64|       563.0|           28.15|\n",
      "|   7326.092|          6462.462|             25.45|           16|     129249.24|             502.2414|           576.87787|        2545.0|         50224.14|       57687.785|       841.0|           42.05|\n",
      "|   3555.235|          7226.672|              25.4|           20|     144533.44|             372.3458|           417.08722|        2540.0|         37234.58|       41708.723|       821.0|           41.05|\n",
      "|  910.39325|         3189.8455|             25.91|           40|      63796.91|            226.29305|            199.3462|        2591.0|        22629.305|       19934.621|      1266.0|            63.3|\n",
      "|  12474.419|          9405.213|             25.78|            9|     188104.27|             881.2983|           888.29443|        2578.0|         88129.83|       88829.445|       687.0|           34.35|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "val features1 = List(\"team_budget\", \"average_attendance\", \"average_player_age\", \"team_position\", \n",
    "                     \"sum_attendance\", \"average_player_rating\", \"average_player_worth\", \"sum_player_age\", \n",
    "                     \"sum_player_rating\", \"sum_player_worth\", \"sum_position\", \"average_position\")\n",
    "Hops.getFeatures(features1).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also explicitly specify the optional arguments: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+\n",
      "|team_budget|score|\n",
      "+-----------+-----+\n",
      "|  11296.577|    1|\n",
      "|   4969.735|    3|\n",
      "|  21319.533|    2|\n",
      "|  15072.062|    1|\n",
      "|  12957.076|    3|\n",
      "+-----------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.queryFeaturestore(\n",
    "    \"SELECT team_budget, score \" +\n",
    "    \"FROM teams_features_1 JOIN games_features_1 ON \" +\n",
    "    \"games_features_1.home_team_id = teams_features_1.team_id\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Free Text SQL Query from the Feature Store\n",
    "\n",
    "For complex queries that cannot be inferred by the helper functions, enter the sql directly to the method `featurestore.sql()` it will default to the project specific feature store but you can also specify it explicitly. If you are proficient in SQL, this is the most efficient and preferred way to query the feature store."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+-------------+\n",
      "|team_budget|team_id|team_position|\n",
      "+-----------+-------+-------------+\n",
      "|  12957.076|      1|            1|\n",
      "|  2403.3704|      2|            2|\n",
      "|  3390.3755|      3|            3|\n",
      "|  13547.429|      4|            4|\n",
      "+-----------+-------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.queryFeaturestore(\"SELECT * FROM teams_features_1 WHERE team_position < 5\").read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Writing to the Feature Store"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating New Feature Groups\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets create a new featuregroup called **teams_features_spanish** that contains the same contents as the feature group teams_features except the the columns are renamed to spanish"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "teamsFeaturesDf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [team_budget: float, team_id: int ... 1 more field]\n"
     ]
    }
   ],
   "source": [
    "val teamsFeaturesDf = Hops.getFeaturegroup(\"teams_features\").read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "teamsFeaturesDf2: org.apache.spark.sql.DataFrame = [equipo_presupuesto: float, equipo_id: int ... 1 more field]\n"
     ]
    }
   ],
   "source": [
    "val teamsFeaturesDf2 = teamsFeaturesDf.withColumnRenamed(\n",
    "    \"team_id\", \"equipo_id\").withColumnRenamed(\n",
    "    \"team_budget\", \"equipo_presupuesto\").withColumnRenamed(\n",
    "    \"team_position\", \"equipo_posicion\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---------+---------------+\n",
      "|equipo_presupuesto|equipo_id|equipo_posicion|\n",
      "+------------------+---------+---------------+\n",
      "|         12957.076|        1|              1|\n",
      "|         2403.3704|        2|              2|\n",
      "|         3390.3755|        3|              3|\n",
      "|         13547.429|        4|              4|\n",
      "|          9678.333|        5|              5|\n",
      "+------------------+---------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "teamsFeaturesDf2.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets now create a new featuregroup using the transformed dataframe (we'll explain the statistics part later on in this notebook)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.createFeaturegroup(\"teams_features_spanish\").setDataframe(teamsFeaturesDf2).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also explicitly specify optional arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.createFeaturegroup(\"teams_features_spanish\").setDataframe(teamsFeaturesDf2).setFeaturestore(Hops.getProjectFeaturestore.read).setDescriptiveStats(false).setFeatureCorr(false).setFeatureHistograms(false).setClusterAnalysis(false).setStatColumns(List[String]().asJava).setNumBins(5).setCorrMethod(\"pearson\").setDescription(\"a spanish version of teams_features\").setVersion(2).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By default the new featuregroup will be created in the project's featurestore and the statistics for the new featuregroup will be computed based on the provided spark dataframe. You can configure this behaviour by modifying the default arguments and filling in extra metadata."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Create an On-Demand Feature Group\n",
    "\n",
    "Feature Groups in Hopsworks can be of two types: \n",
    "\n",
    "- **Cached Feature Groups** are pre-computed and stored inside Hopsworks as Hive tables for historical data and MySQL Cluster tables for online data.\n",
    "- **On-Demand Feature Groups** are computed on-demand using a JDBC connector and a SQL query. On-Demand Feature Groups can be stored in any JDBC-compliant data store.\n",
    "\n",
    "To create an on-demand feature group, you must first configure a storage connector ti the JDBC backend that you want to query and then you can use the method `createFeaturegroup().setOnDemand(true)` to create the on-demand feature group.\n",
    "\n",
    "By default, a Feature Store in Hopsworks has two default JDBC storage connectors configured:\n",
    "\n",
    "- `project_name`: a storage connector for the Hive database of the project\n",
    "- `project_name_featurestore`: a storage connector for the Feature Store database of the project"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can list the available storage connectors in the feature store, using the method `getStorageConnectors()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res17: java.util.List[String] = [demo_featurestore_admin000_featurestore, demo_featurestore_admin000, demo_featurestore_admin000_meb1_onlinefeaturestore, demo_featurestore_admin000_Training_Datasets]\n"
     ]
    }
   ],
   "source": [
    "Hops.getStorageConnectors().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we inspect the Feature Store hive database, we can see that there is a table called `games_features_1`, lets use the JDBC connector called `<project_name>_featurestore` and the SQL string `\"SELECT * FROM games_features_1 WHERE score > 1\"` to create an on-demand feature group called `games_features_on_demand`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----------+\n",
      "|            database|           tableName|isTemporary|\n",
      "+--------------------+--------------------+-----------+\n",
      "|demo_featurestore...|attendances_featu...|      false|\n",
      "|demo_featurestore...|    games_features_1|      false|\n",
      "|demo_featurestore...|games_features_hu...|      false|\n",
      "|demo_featurestore...|houses_for_sale_f...|      false|\n",
      "|demo_featurestore...|  players_features_1|      false|\n",
      "+--------------------+--------------------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.queryFeaturestore(\"show tables\").read().show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+---------+-------+\n",
      "|    col_name|data_type|comment|\n",
      "+------------+---------+-------+\n",
      "|away_team_id|      int|      -|\n",
      "|home_team_id|      int|      -|\n",
      "|       score|      int|      -|\n",
      "+------------+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.queryFeaturestore(\"describe games_features_1\").read().show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "featuregroupName: String = games_features_on_demand\n",
      "storageConnector: String = demo_featurestore_admin000_featurestore\n",
      "query: String = SELECT * FROM games_features_1 WHERE score > 1\n"
     ]
    }
   ],
   "source": [
    "val featuregroupName = \"games_features_on_demand\"\n",
    "val storageConnector = \"demo_featurestore_admin000_featurestore\"\n",
    "val query = \"SELECT * FROM games_features_1 WHERE score > 1\"\n",
    "Hops.createFeaturegroup(featuregroupName).setOnDemand(true).setJdbcConnector(storageConnector).setSqlQuery(query).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On-Demand Feature Groups can be queried just like cached feature groups. When you query an on-demand feature group, it will open a JDBC connection and apply the associated SQL query and return the resulting dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+------------+-----+\n",
      "|away_team_id|home_team_id|score|\n",
      "+------------+------------+-----+\n",
      "|          46|          32|    3|\n",
      "|          15|          13|    2|\n",
      "|          50|           1|    3|\n",
      "|          36|          39|    2|\n",
      "|          14|           8|    3|\n",
      "+------------+------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"games_features_on_demand\").read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets create an on-demand feature group with the same JDBC connector and the table `players_features_1`, and see how we can make queries that join on-demand feature groups with cached feature groups dynamically."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "query: String = SELECT average_player_age as average_player_age_on_dmd, average_player_worth as average_player_worth_on_dmd, team_id FROM players_features_1\n",
      "storageConnector: String = demo_featurestore_admin000_featurestore\n",
      "featuregroupName: String = players_features_on_demand\n"
     ]
    }
   ],
   "source": [
    "val query = \"SELECT average_player_age as average_player_age_on_dmd, average_player_worth as average_player_worth_on_dmd, team_id FROM players_features_1\"\n",
    "val storageConnector = Hops.getProjectName() + \"_featurestore\"\n",
    "val featuregroupName = \"players_features_on_demand\"\n",
    "Hops.createFeaturegroup(featuregroupName).setOnDemand(true).setJdbcConnector(storageConnector).setSqlQuery(query).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When you make a query to the feature store that cross multiple feature groups, including on-demand feature groups. The Query Planner that comes with the Python SDK will first fetch the on-demand feature groups and then register them as SparkSQL temporary tables, before joining them with the features from the cached feature groups.\n",
    "\n",
    "Note that when querying on-demand-featuregroups, you typically have to supply the `joinKey` and providing a map of Map with entries of `<feature group name> -> <feature group version>` manually as the query planner will often not have information about the columns in the on-demand feature groups and hence cannot infer the join key or infer in which on-demand feature group a certain feature exists."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "features: List[String] = List(average_player_age_on_dmd, average_player_worth_on_dmd, average_attendance, sum_attendance)\n",
      "featuregroupsMap: scala.collection.immutable.Map[String,Integer] = Map(players_features_on_demand -> 1, attendances_features -> 1)\n",
      "javaFeaturegroupsMap: java.util.HashMap[String,Integer] = {players_features_on_demand=1, attendances_features=1}\n",
      "+-------------------------+---------------------------+------------------+--------------+\n",
      "|average_player_age_on_dmd|average_player_worth_on_dmd|average_attendance|sum_attendance|\n",
      "+-------------------------+---------------------------+------------------+--------------+\n",
      "|                    24.63|                   231.8708|         3587.5015|      71750.03|\n",
      "|                    25.71|                  223.71338|         2532.1638|     50643.277|\n",
      "|                    25.63|                  280.11465|         3397.8066|      67956.13|\n",
      "|                    25.65|                  307.87268|          3271.934|      65438.68|\n",
      "|                     25.5|                  298.78235|         4074.8047|      81496.09|\n",
      "+-------------------------+---------------------------+------------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "val features = List(\"average_player_age_on_dmd\", \"average_player_worth_on_dmd\", \"average_attendance\", \"sum_attendance\")\n",
    "val featuregroupsMap = Map[String, Integer](\n",
    "    \"players_features_on_demand\"->1,\n",
    "    \"attendances_features\"->1\n",
    ")\n",
    "val javaFeaturegroupsMap = new java.util.HashMap[String, Integer](featuregroupsMap)\n",
    "Hops.getFeatures(features).setJoinKey(\"team_id\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setFeaturegroupsAndVersions(javaFeaturegroupsMap).read().show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|average_player_age_on_dmd|\n",
      "+-------------------------+\n",
      "|                    24.63|\n",
      "|                    25.71|\n",
      "|                    25.63|\n",
      "|                    25.65|\n",
      "|                     25.5|\n",
      "+-------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeature(\"average_player_age_on_dmd\").setFeaturegroup(\"players_features_on_demand\").read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a Hudi Feature Group\n",
    "\n",
    "TLDR; Hudi is a storage abstraction/library build on top of Spark. A Hudi dataset stores data in Parquet files and maintains additional metadata to make upserts efficient. A Hudi ingest job is intended to be run as a streaming ingest job, on an interval such as every 15 minutes, reading deltas from a message-bus like Kafka and ingesting the deltas incrementally into a data lake.\n",
    "\n",
    "Hudi also makes it possible to **time-travel** by inspecting datasets at different commit times, which is especially useful for a feature store as it allows us to make ML experiments reproducible and easily be able to generate training data by backfilling labels."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "import org.apache.hadoop.fs.FileSystem\n",
      "import org.apache.hudi.DataSourceReadOptions\n",
      "import org.apache.hudi.DataSourceWriteOptions\n",
      "import org.apache.hudi.HoodieDataSourceHelpers\n",
      "import org.apache.hudi.NonpartitionedKeyGenerator\n",
      "import org.apache.hudi.SimpleKeyGenerator\n",
      "import org.apache.hudi.common.model.HoodieTableType\n",
      "import org.apache.hudi.config.HoodieWriteConfig\n",
      "import org.apache.hudi.hive.MultiPartKeysValueExtractor\n",
      "import org.apache.hudi.hive.NonPartitionedExtractor\n",
      "import org.apache.log4j.LogManager\n",
      "import org.apache.log4j.Logger\n",
      "import org.apache.spark.api.java.JavaSparkContext\n",
      "import org.apache.spark.sql.DataFrameWriter\n",
      "import org.apache.spark.sql.Dataset\n",
      "import org.apache.spark.sql.Row\n",
      "import org.apache.spark.sql.SaveMode\n",
      "import org.apache.spark.sql.SparkSession\n",
      "import io.hops.util.Hops\n",
      "import org.apache.spark.sql._\n",
      "import spark.implicits._\n",
      "import org.apache.spark.sql.types._\n",
      "import java.sql.Date\n",
      "import java.sql.Timestamp\n",
      "import org.apache.hadoop.fs.{FileSystem, Path}\n"
     ]
    }
   ],
   "source": [
    "import org.apache.hadoop.fs.FileSystem;\n",
    "import org.apache.hudi.DataSourceReadOptions;\n",
    "import org.apache.hudi.DataSourceWriteOptions;\n",
    "import org.apache.hudi.HoodieDataSourceHelpers;\n",
    "import org.apache.hudi.NonpartitionedKeyGenerator;\n",
    "import org.apache.hudi.SimpleKeyGenerator;\n",
    "import org.apache.hudi.common.model.HoodieTableType;\n",
    "import org.apache.hudi.config.HoodieWriteConfig;\n",
    "import org.apache.hudi.hive.MultiPartKeysValueExtractor;\n",
    "import org.apache.hudi.hive.NonPartitionedExtractor;\n",
    "import org.apache.log4j.LogManager;\n",
    "import org.apache.log4j.Logger;\n",
    "import org.apache.spark.api.java.JavaSparkContext;\n",
    "import org.apache.spark.sql.DataFrameWriter;\n",
    "import org.apache.spark.sql.Dataset;\n",
    "import org.apache.spark.sql.Row;\n",
    "import org.apache.spark.sql.SaveMode;\n",
    "import org.apache.spark.sql.SparkSession;\n",
    "import io.hops.util.Hops\n",
    "import org.apache.spark.sql._\n",
    "import spark.implicits._\n",
    "import org.apache.spark.sql.types._\n",
    "import java.sql.Date\n",
    "import java.sql.Timestamp\n",
    "import org.apache.hadoop.fs.{FileSystem, Path}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sampleData: Seq[org.apache.spark.sql.Row] = List([1,2019-03-02,0.4151,Sweden], [2,2019-05-01,1.2151,Ireland], [3,2019-08-06,0.2151,Belgium], [4,2019-08-06,0.8151,Russia])\n",
      "schema: List[org.apache.spark.sql.types.StructField] = List(StructField(id,IntegerType,true), StructField(date,DateType,true), StructField(value,FloatType,true), StructField(country,StringType,true))\n",
      "sampleDf: org.apache.spark.sql.DataFrame = [id: int, date: date ... 2 more fields]\n",
      "partitionCols: List[String] = List(date)\n"
     ]
    }
   ],
   "source": [
    "val sampleData = Seq(\n",
    "    Row(1, Date.valueOf(\"2019-02-30\"), 0.4151f, \"Sweden\"),\n",
    "    Row(2, Date.valueOf(\"2019-05-01\"), 1.2151f, \"Ireland\"),\n",
    "    Row(3, Date.valueOf(\"2019-08-06\"), 0.2151f, \"Belgium\"),\n",
    "    Row(4, Date.valueOf(\"2019-08-06\"), 0.8151f, \"Russia\")\n",
    ")\n",
    "val schema = \n",
    " scala.collection.immutable.List(\n",
    "  StructField(\"id\", IntegerType, true),\n",
    "  StructField(\"date\", DateType, true),\n",
    "  StructField(\"value\", FloatType, true),\n",
    "  StructField(\"country\", StringType, true) \n",
    ")\n",
    "val sampleDf = spark.createDataFrame(\n",
    "  spark.sparkContext.parallelize(sampleData),\n",
    "  StructType(schema)\n",
    ")\n",
    "val partitionCols = List(\"date\")\n",
    "(Hops.createFeaturegroup(\"hudi_featuregroup_test_tour\")\n",
    "                         .setHudi(true)\n",
    "                         .setPartitionBy(partitionCols)\n",
    "                         .setDataframe(sampleDf)\n",
    "                         .setPrimaryKey(List(\"id\")).write())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+------------------+----------------------+--------------------+---+------+-------+-------------+\n",
      "|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name| id| value|country|         date|\n",
      "+-------------------+--------------------+------------------+----------------------+--------------------+---+------+-------+-------------+\n",
      "|     20191112151814|  20191112151814_2_3|                 3|         1565049600000|c3d12cef-b730-473...|  3|0.2151|Belgium|1565049600000|\n",
      "|     20191112151814|  20191112151814_1_2|                 2|         1556668800000|531d07ce-374c-4f7...|  2|1.2151|Ireland|1556668800000|\n",
      "|     20191112151814|  20191112151814_3_4|                 4|         1565049600000|d243a4f5-5831-496...|  4|0.8151| Russia|1565049600000|\n",
      "|     20191112151814|  20191112151814_0_1|                 1|         1551484800000|0e97f777-b6eb-462...|  1|0.4151| Sweden|1551484800000|\n",
      "+-------------------+--------------------+------------------+----------------------+--------------------+---+------+-------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"hudi_featuregroup_test_tour\").read.show(20)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also override Hudi specific arguments using the setHudiArgs and setHudiBasePath methods:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hudiArgs: scala.collection.immutable.Map[String,String] = Map(hoodie.datasource.write.payload.class -> org.apache.hudi.OverwriteWithLatestAvroPayload)\n"
     ]
    }
   ],
   "source": [
    "val hudiArgs = Map[String, String](\n",
    "    \"hoodie.datasource.write.payload.class\"-> \"org.apache.hudi.OverwriteWithLatestAvroPayload\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "(Hops.createFeaturegroup(\"hudi_featuregroup_test_tour_second\")\n",
    "                         .setHudi(true)\n",
    "                         .setPartitionBy(partitionCols)\n",
    "                         .setDataframe(sampleDf)\n",
    "                         .setHudiBasePath(s\"hdfs:///Projects/${Hops.getProjectName}/Resources/hudi_featuregroup_test_tour_second\")\n",
    "                         .setHudiArgs(hudiArgs)\n",
    "                         .setPrimaryKey(List(\"id\")).write())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+------------------+----------------------+--------------------+---+------+-------+-------------+\n",
      "|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name| id| value|country|         date|\n",
      "+-------------------+--------------------+------------------+----------------------+--------------------+---+------+-------+-------------+\n",
      "|     20191112151835|  20191112151835_1_6|                 2|         1556668800000|de23cbaa-f407-451...|  2|1.2151|Ireland|1556668800000|\n",
      "|     20191112151835|  20191112151835_2_7|                 3|         1565049600000|ab98dda5-5448-496...|  3|0.2151|Belgium|1565049600000|\n",
      "|     20191112151835|  20191112151835_0_5|                 1|         1551484800000|747a83ac-2534-445...|  1|0.4151| Sweden|1551484800000|\n",
      "|     20191112151835|  20191112151835_3_8|                 4|         1565049600000|ef35bf0d-85b8-435...|  4|0.8151| Russia|1565049600000|\n",
      "+-------------------+--------------------+------------------+----------------------+--------------------+---+------+-------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"hudi_featuregroup_test_tour_second\").read.show(20)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also utilize the `insertIntoFeaturegroup` wrapper to make Upserts into Hudi datasets:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "import scala.collection.JavaConversions._\n",
      "import collection.JavaConverters._\n",
      "upsertData: Seq[org.apache.spark.sql.Row] = List([5,2019-03-02,0.7921,Northern Ireland], [1,2019-05-01,1.151,Norway], [3,2019-08-06,0.999,Belgium], [6,2019-08-06,0.0151,France])\n",
      "schema: List[org.apache.spark.sql.types.StructField] = List(StructField(id,IntegerType,true), StructField(date,DateType,true), StructField(value,FloatType,true), StructField(country,StringType,true))\n",
      "upsertDf: org.apache.spark.sql.DataFrame = [id: int, date: date ... 2 more fields]\n",
      "partitionCols: List[String] = List(date)\n"
     ]
    }
   ],
   "source": [
    "import scala.collection.JavaConversions._\n",
    "import collection.JavaConverters._\n",
    "val upsertData = Seq(\n",
    "    Row(5, Date.valueOf(\"2019-02-30\"), 0.7921f, \"Northern Ireland\"), //Insert\n",
    "    Row(1, Date.valueOf(\"2019-05-01\"), 1.151f, \"Norway\"), //Update\n",
    "    Row(3, Date.valueOf(\"2019-08-06\"), 0.999f, \"Belgium\"), //Update\n",
    "    Row(6, Date.valueOf(\"2019-08-06\"), 0.0151f, \"France\") //Insert\n",
    ")\n",
    "val schema = \n",
    " scala.collection.immutable.List(\n",
    "  StructField(\"id\", IntegerType, true),\n",
    "  StructField(\"date\", DateType, true),\n",
    "  StructField(\"value\", FloatType, true),\n",
    "  StructField(\"country\", StringType, true) \n",
    ")\n",
    "val upsertDf = spark.createDataFrame(\n",
    "  spark.sparkContext.parallelize(upsertData),\n",
    "  StructType(schema)\n",
    ")\n",
    "val partitionCols = List(\"date\")\n",
    "(Hops.insertIntoFeaturegroup(\"hudi_featuregroup_test_tour\")\n",
    "                         .setPartitionBy(partitionCols)\n",
    "                         .setDataframe(upsertDf)\n",
    "                         .setMode(\"append\")\n",
    "                         .setPrimaryKey(List(\"id\")).write())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "timeline: org.apache.hudi.common.table.HoodieTimeline = org.apache.hudi.common.table.timeline.HoodieDefaultTimeline: [20191112151814__commit__COMPLETED],[20191112151859__commit__COMPLETED]\n"
     ]
    }
   ],
   "source": [
    "val timeline = HoodieDataSourceHelpers.allCompletedCommitsCompactions(FileSystem.get(sc.hadoopConfiguration), \n",
    "                                     s\"hdfs:///Projects/${Hops.getProjectName}/Resources/hudi_featuregroup_test_tour_1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "firstTimestamp: String = 20191112151814\n"
     ]
    }
   ],
   "source": [
    "val firstTimestamp = timeline.firstInstant.get.getTimestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "secondTimestamp: String = 20191112151859\n"
     ]
    }
   ],
   "source": [
    "val secondTimestamp = timeline.nthInstant(1).get.getTimestamp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Synchronize a Hive Table with the Feature Store\n",
    "\n",
    "The Feature Store SDK supports method for synchronizing existing Hive tables with the feature store, using `syncHiveTableWithFeaturestore`\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create an example Hive Table (praxis in the feature store is to have `_version` suffix on the table):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "import scala.collection.JavaConversions._\n",
      "import collection.JavaConverters._\n",
      "sampleData: Seq[org.apache.spark.sql.Row] = List([1,2019-03-02,0.4151,Sweden], [2,2019-05-01,1.2151,Ireland], [3,2019-08-06,0.2151,Belgium], [4,2019-08-06,0.8151,Russia])\n",
      "schema: List[org.apache.spark.sql.types.StructField] = List(StructField(id,IntegerType,true), StructField(date,DateType,true), StructField(value,FloatType,true), StructField(country,StringType,true))\n",
      "sampleDf: org.apache.spark.sql.DataFrame = [id: int, date: date ... 2 more fields]\n",
      "res28: org.apache.spark.sql.DataFrame = []\n"
     ]
    }
   ],
   "source": [
    "import scala.collection.JavaConversions._\n",
    "import collection.JavaConverters._\n",
    "val sampleData = Seq(\n",
    "    Row(1, Date.valueOf(\"2019-02-30\"), 0.4151f, \"Sweden\"),\n",
    "    Row(2, Date.valueOf(\"2019-05-01\"), 1.2151f, \"Ireland\"),\n",
    "    Row(3, Date.valueOf(\"2019-08-06\"), 0.2151f, \"Belgium\"),\n",
    "    Row(4, Date.valueOf(\"2019-08-06\"), 0.8151f, \"Russia\")\n",
    ")\n",
    "val schema = \n",
    " scala.collection.immutable.List(\n",
    "  StructField(\"id\", IntegerType, true),\n",
    "  StructField(\"date\", DateType, true),\n",
    "  StructField(\"value\", FloatType, true),\n",
    "  StructField(\"country\", StringType, true) \n",
    ")\n",
    "val sampleDf = spark.createDataFrame(\n",
    "  spark.sparkContext.parallelize(sampleData),\n",
    "  StructType(schema)\n",
    ")\n",
    "spark.sql(s\"use ${Hops.getProjectFeaturestore.read}\")\n",
    "sampleDf.write.mode(\"overwrite\").saveAsTable(\"hive_fs_sync_example_1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------------------+------------------------------------+-----------+\n",
      "|database                               |tableName                           |isTemporary|\n",
      "+---------------------------------------+------------------------------------+-----------+\n",
      "|demo_featurestore_admin000_featurestore|attendances_features_1              |false      |\n",
      "|demo_featurestore_admin000_featurestore|games_features_1                    |false      |\n",
      "|demo_featurestore_admin000_featurestore|games_features_hudi_tour_1          |false      |\n",
      "|demo_featurestore_admin000_featurestore|hive_fs_sync_example_1              |false      |\n",
      "|demo_featurestore_admin000_featurestore|houses_for_sale_featuregroup_3      |false      |\n",
      "|demo_featurestore_admin000_featurestore|hudi_featuregroup_test_tour_1       |false      |\n",
      "|demo_featurestore_admin000_featurestore|hudi_featuregroup_test_tour_second_1|false      |\n",
      "|demo_featurestore_admin000_featurestore|players_features_1                  |false      |\n",
      "|demo_featurestore_admin000_featurestore|season_scores_features_1            |false      |\n",
      "|demo_featurestore_admin000_featurestore|teams_features_1                    |false      |\n",
      "|demo_featurestore_admin000_featurestore|teams_features_spanish_1            |false      |\n",
      "|demo_featurestore_admin000_featurestore|teams_features_spanish_2            |false      |\n",
      "+---------------------------------------+------------------------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"show tables\").show(20, false)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can synchronize the Hive table with the feature store (so that it shows up in the Feature store UI and so that we can query the table using the Feature Store API). The SyncTool will look for a Hive table callled `name_version`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.syncHiveTableWithFeaturestore(\"hive_fs_sync_example\").setVersion(1).setDescription(\"Hive Sync Test\").write"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res32: java.util.List[String] = [teams_features_1, games_features_on_demand_tour_1, players_features_1, games_features_hudi_tour_1, games_features_1, attendances_features_1, season_scores_features_1, teams_features_spanish_1, players_features_on_demand_1, hudi_featuregroup_test_tour_1, hudi_featuregroup_test_tour_second_1, hive_fs_sync_example_1, hudi_featuregroup_test_tour_1, teams_features_spanish_2, games_features_on_demand_1]\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroups().read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+------+-------+\n",
      "| id|      date| value|country|\n",
      "+---+----------+------+-------+\n",
      "|  3|2019-08-06|0.2151|Belgium|\n",
      "|  4|2019-08-06|0.8151| Russia|\n",
      "|  1|2019-03-02|0.4151| Sweden|\n",
      "|  2|2019-05-01|1.2151|Ireland|\n",
      "+---+----------+------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"hive_fs_sync_example\").read.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Create a New Version of A Feature Group\n",
    "\n",
    "To create a new version, simply use the `createFeaturegroup` method and change the version argument:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.createFeaturegroup(\"teams_features_spanish\").setDataframe(teamsFeaturesDf2).setVersion(3).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get the Latest Version of a Feature Group (0 if no version exist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "latestVersion: Integer = 3\n",
      "res35: Integer = 3\n"
     ]
    }
   ],
   "source": [
    "val latestVersion = Hops.getLatestFeaturegroupVersion(\"teams_features_spanish\").read()\n",
    "latestVersion"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also specify arguments explicitly:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res36: Integer = 3\n"
     ]
    }
   ],
   "source": [
    "Hops.getLatestFeaturegroupVersion(\"teams_features_spanish\").setFeaturestore(Hops.getProjectFeaturestore.read).read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Updating the Metadata Cache\n",
    "\n",
    "By default the metadata of the feature store will be cached on the client-side after the first query with the feature store, the metadata is used by most queries to the store to figure out how to fetch the features etc. To update the metadata use the following API method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateFeaturestoreMetadataCache().setFeaturestore(Hops.getProjectFeaturestore.read).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Featuregroup Partitioning \n",
    "\n",
    "Featuregroups are stored as Hive tables, meaning that they can be partitioned for improved **read-query** performance. We use dynamic partitioning where the partition keys are specified on creation of a featuregroup. To set the partitions, simply specify the `partitionBy` argument to `createFeaturegroup()`. The `partition_by` argument should be set as a list of the columns that you want to partition the table on, see examples below.\n",
    "\n",
    "Partitioning is not supported for training datasets as those are meant to be immutable blobs used for training (e.g petastorm or tfrecords), and do not need to be optimized for query performance. \n",
    "\n",
    "Feature groups in Hive might need to be read-query optimized however, since it might be used to create a lot of different training datasets, using different subsets of features."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As an example we can take the `games_features` featuregroup created by the featurestore tour and re-create it as a new featuregroup called `games_features_partitioned` where we partition on the `score` column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "gamesFeaturesDf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [away_team_id: int, home_team_id: int ... 1 more field]\n"
     ]
    }
   ],
   "source": [
    "val gamesFeaturesDf = Hops.getFeaturegroup(\"games_features\").read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "partitionCols: List[String] = List(score)\n"
     ]
    }
   ],
   "source": [
    "val partitionCols = List(\"score\")\n",
    "Hops.createFeaturegroup(\"games_features_partitioned\").setPartitionBy(partitionCols).setDataframe(gamesFeaturesDf).setDescriptiveStats(false).setFeatureCorr(false).setFeatureHistograms(false).setClusterAnalysis(false).setDescription(\"games_features partitioned by score\").write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also partition on multiple columns (watch out so that the number of partitions don't get too many though). Also note that it is not allowed to partition on the primary key, which does not make sense either since the primary key should be unique. If you try to partition on the primary key, the partitioning will simply be skipped.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "partitionCols: List[String] = List(score, home_team_id)\n"
     ]
    }
   ],
   "source": [
    "val partitionCols = List(\"score\", \"home_team_id\")\n",
    "Hops.createFeaturegroup(\"games_features_double_partitioned\").setPartitionBy(partitionCols).setDataframe(gamesFeaturesDf).setDescriptiveStats(false).setFeatureCorr(false).setFeatureHistograms(false).setClusterAnalysis(false).setDescription(\"games_features partitioned by score and away_team_id\").write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To verify the partitions we can use the utility function `getFeaturegroupPartitions`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|partition|\n",
      "+---------+\n",
      "|  score=1|\n",
      "|  score=2|\n",
      "|  score=3|\n",
      "+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroupPartitions(\"games_features_partitioned\").read().show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|           partition|\n",
      "+--------------------+\n",
      "|home_team_id=1/sc...|\n",
      "|home_team_id=1/sc...|\n",
      "|home_team_id=10/s...|\n",
      "|home_team_id=10/s...|\n",
      "|home_team_id=10/s...|\n",
      "|home_team_id=11/s...|\n",
      "|home_team_id=11/s...|\n",
      "|home_team_id=11/s...|\n",
      "|home_team_id=12/s...|\n",
      "|home_team_id=12/s...|\n",
      "+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroupPartitions(\"games_features_double_partitioned\").read().show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inserting Into Existing Feature Groups\n",
    "\n",
    "A best practice when working with features in HopsML is to first figure out a model of feature groups and create them  using the Feature Registry UI. This will prepare the feature group schema and create the Hive tables. Once the empty feature groups are created, then you can insert into these tables directly."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets first get some sample data to insert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "import org.apache.spark.sql._\n",
      "import spark.implicits._\n",
      "import org.apache.spark.sql.types._\n",
      "sampleData: Seq[org.apache.spark.sql.Row] = List([999,41251.52,1], [998,1319.4,8], [997,21219.1,2])\n",
      "sampleSchema: List[org.apache.spark.sql.types.StructField] = List(StructField(equipo_id,IntegerType,true), StructField(equipo_presupuesto,FloatType,true), StructField(equipo_posicion,IntegerType,true))\n",
      "sampleDF: org.apache.spark.sql.DataFrame = [equipo_id: int, equipo_presupuesto: float ... 1 more field]\n"
     ]
    }
   ],
   "source": [
    "import org.apache.spark.sql._\n",
    "import spark.implicits._\n",
    "import org.apache.spark.sql.types._\n",
    "val sampleData = Seq(\n",
    "    Row(999, 41251.52f, 1),\n",
    "    Row(998, 1319.4f, 8),\n",
    "    Row(997, 21219.1f, 2)\n",
    ")\n",
    "val sampleSchema = List(\n",
    "  StructField(\"equipo_id\", IntegerType, true),\n",
    "  StructField(\"equipo_presupuesto\", FloatType, true),\n",
    "  StructField(\"equipo_posicion\", IntegerType, true)\n",
    ")\n",
    "val sampleDF = spark.createDataFrame(\n",
    "  spark.sparkContext.parallelize(sampleData),\n",
    "  StructType(sampleSchema)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------------------+---------------+\n",
      "|equipo_id|equipo_presupuesto|equipo_posicion|\n",
      "+---------+------------------+---------------+\n",
      "|      999|          41251.52|              1|\n",
      "|      998|            1319.4|              8|\n",
      "|      997|           21219.1|              2|\n",
      "+---------+------------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sampleDF.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res43: Long = 3\n"
     ]
    }
   ],
   "source": [
    "sampleDF.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets inspect the contents of the featuregroup `teams_features_spanish` that we are going to insert the sample data into:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "spanishTeamsFeaturesDf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [equipo_presupuesto: float, equipo_id: int ... 1 more field]\n"
     ]
    }
   ],
   "source": [
    "val spanishTeamsFeaturesDf = Hops.getFeaturegroup(\"teams_features_spanish\").read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---------+---------------+\n",
      "|equipo_presupuesto|equipo_id|equipo_posicion|\n",
      "+------------------+---------+---------------+\n",
      "|         12957.076|        1|              1|\n",
      "|         2403.3704|        2|              2|\n",
      "|         3390.3755|        3|              3|\n",
      "|         13547.429|        4|              4|\n",
      "|          9678.333|        5|              5|\n",
      "+------------------+---------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spanishTeamsFeaturesDf.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res45: Long = 50\n"
     ]
    }
   ],
   "source": [
    "spanishTeamsFeaturesDf.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can insert the sample data and verify the new contents of the featuregroup. By default the insert mode is \"append\", the featurestore is the project's featurestore, the version is 1 and statistics will be updated based on the previously made settings when the featuregroup was created (we cover statistics later on in this notebook)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.insertIntoFeaturegroup(\"teams_features_spanish\").setDataframe(sampleDF).setMode(\"append\").write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also explicitly specify the optional arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.insertIntoFeaturegroup(\"teams_features_spanish\").setDataframe(sampleDF).setMode(\"append\").setVersion(1).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets fetch the updated feature group from the feature store and verify that the update was successful"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "spanishTeamsFeaturesUpdatedDf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [equipo_presupuesto: float, equipo_id: int ... 1 more field]\n"
     ]
    }
   ],
   "source": [
    "val spanishTeamsFeaturesUpdatedDf = Hops.getFeaturegroup(\"teams_features_spanish\").read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---------+---------------+\n",
      "|equipo_presupuesto|equipo_id|equipo_posicion|\n",
      "+------------------+---------+---------------+\n",
      "|         12957.076|        1|              1|\n",
      "|         2403.3704|        2|              2|\n",
      "|         3390.3755|        3|              3|\n",
      "|         13547.429|        4|              4|\n",
      "|          9678.333|        5|              5|\n",
      "+------------------+---------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spanishTeamsFeaturesUpdatedDf.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res49: Long = 56\n"
     ]
    }
   ],
   "source": [
    "spanishTeamsFeaturesUpdatedDf.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The two supported insert modes are \"append\" and \"overwrite\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.insertIntoFeaturegroup(\"teams_features_spanish\").setDataframe(sampleDF).setMode(\"overwrite\").write()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------------+------------------+\n",
      "|equipo_id|equipo_posicion|equipo_presupuesto|\n",
      "+---------+---------------+------------------+\n",
      "|      999|          41251|               1.0|\n",
      "|      998|           1319|               8.0|\n",
      "|      997|          21219|               2.0|\n",
      "+---------+---------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"teams_features_spanish\").read().show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res52: Long = 3\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroup(\"teams_features_spanish\").read().count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import External Feature Datasets to the Feature Store\n",
    "\n",
    "Feature datasets stored externally to Hopsworks can be imported using the `importFeaturegroup` operation in the Featurestore SDK. For example, say that you have a dataset stored on S3 that you want to import. To import the dataset from S3, first configure an s3 connector for your feature store, e.g with name `my_s3_connector`. The connector is used for authenticating with S3 and for specifying your bucket name. Once the connector is configured, you can use the `importFeaturegroup` operation as follows:\n",
    "```scala\n",
    "Hops.importFeaturegroup(featuregroup_name)\n",
    "    .setExternalPath(s3_bucket_path)\n",
    "    .setStorageConnector(\"my_s3_connector\")\n",
    "    .setDataFormat(\"tfrecords\")\n",
    "    .write\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Group Statistics\n",
    "\n",
    "Statistics about a featuregroup can be useful in the stage of feature engineering and when deciding which features to use for training. If statistics have been computed for a feature group, it can be viewed in the Hopsworks Feature Registry UI. \n",
    "\n",
    "This is particularly useful within large organizations where data scientists from different teams can re-use and explore new features by browsing features in the feature store and analyzing the statistics.\n",
    "\n",
    "![Feature Registry Statistics Visualization](./images/fg_stats_1.png \"Feature Registry Statistics Visualization\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you might have notived earlier in this notebook, the `createFeaturegroup` method has arguments for configuring the statistics to be computed as data is added. These settings will be saved along with the rest of the meta data about the featuregroup and will be applied when new data is inserted into the featuregroup using the `insertIntoFeaturegroup` method.\n",
    "\n",
    "You can use the `updateFeaturegroupStats()` method to update the settings for the statistics of a feature group and recompute the statistics without inserting any new data. By default it will compute the statistics based on the previous settings, use the project's featurestore and use version 1 of the featuregroup:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateFeaturegroupStats(\"teams_features\").write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also specify the optional arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateFeaturegroupStats(\"teams_features\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setVersion(1).setDescriptiveStats(true).setFeatureCorr(true).setFeatureHistograms(true).setClusterAnalysis(true).setStatColumns(null).setNumBins(10).setCorrMethod(\"pearson\").setNumClusters(5).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you only want to compute statistics for certain set of columns and exclude surrogate key-columns for example, you can use the argument `statColumns` to specify which columns to include:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "statColumns: java.util.List[String] = [team_budget, team_position]\n"
     ]
    }
   ],
   "source": [
    "val statColumns = List[String](\"team_budget\", \"team_position\").asJava\n",
    "Hops.updateFeaturegroupStats(\"teams_features\").setStatColumns(statColumns).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training Datasets\n",
    "\n",
    "To group data in the feature store we use three concepts:\n",
    "\n",
    "- Feature\n",
    "- Feature group\n",
    "- Training Dataset\n",
    "\n",
    "Typically during the feature engineering phase of a machine learning project, you compute a set of features for each type of data that you have, these features are naturally grouped into a documented and versioned **feature group**. \n",
    "\n",
    "In practice, it is common that organizations have many different type of datasets that they can extract features from, for example if you are building a recommendation system you might have demographic data about each user as well as user-activity data. \n",
    "\n",
    "When you train a machine learning model, you want to use all features that have predictive power and that the model can learn from. At this point, we can create a training dataset of features from several different feature groups and use that for training. That is the purpose of the training dataset abstraction. \n",
    "\n",
    "Of course you can always just save a group of features anywhere inside your project, e.g as a csv, or .tfrecords file. However, by using the feature store you can create **managed** training datasets. Managed training datasets will show up in the feature registry UI and will automatically be versioned, documented and reproducible. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Metadata for a training dataset can be created from the Hopsworks UI or directly from the API with the function `createTrainingDataset`. The training datasets in a project are stored in a top-level dataset called `projectName_Training_Datasets`, (i.e `hdfs:///Projects/<ProjectName>/<ProjectName>_Training_Datasets`.\n",
    "\n",
    "Once a training dataset have been created you can find it in the featurestore UI in hopsworks under the tab `Training datasets`, from there you can also edit the metadata if necessary. \n",
    "\n",
    "After a training dataset have been created with the necessary metadata you can save the actual data in the training dataset by using the API function `insertIntoTrainingDataset`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create New Training Dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets create a dataset called `team_position_prediction` by using a set of relevant features from the featurestore. We will combine features from four different feature groups to form this training dataset: `teams_features`, `attendances_features`, `players_features`, `season_scores_features`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Read Features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "features: List[String] = List(team_budget, average_attendance, average_player_age, team_position, sum_attendance, average_player_rating, average_player_worth, sum_player_age, sum_player_rating, sum_player_worth, sum_position, average_position)\n",
      "featuresDf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [team_budget: float, average_attendance: float ... 10 more fields]\n"
     ]
    }
   ],
   "source": [
    "val features = List(\"team_budget\", \"average_attendance\", \"average_player_age\", \"team_position\", \n",
    "                     \"sum_attendance\", \"average_player_rating\", \"average_player_worth\", \"sum_player_age\", \n",
    "                     \"sum_player_rating\", \"sum_player_worth\", \"sum_position\", \"average_position\")\n",
    "val featuresDf = Hops.getFeatures(features).read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|team_budget|average_attendance|average_player_age|team_position|sum_attendance|average_player_rating|average_player_worth|sum_player_age|sum_player_rating|sum_player_worth|sum_position|average_position|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|    7307.94|         19595.973|             24.85|            6|     391919.47|            1311.2384|           1435.2465|        2485.0|        131123.84|       143524.64|       563.0|           28.15|\n",
      "|   7326.092|          6462.462|             25.45|           16|     129249.24|             502.2414|           576.87787|        2545.0|         50224.14|       57687.785|       841.0|           42.05|\n",
      "|   3555.235|          7226.672|              25.4|           20|     144533.44|             372.3458|           417.08722|        2540.0|         37234.58|       41708.723|       821.0|           41.05|\n",
      "|  910.39325|         3189.8455|             25.91|           40|      63796.91|            226.29305|            199.3462|        2591.0|        22629.305|       19934.621|      1266.0|            63.3|\n",
      "|  10290.323|         4964.6475|             26.63|           23|     99292.945|             327.6003|            315.4794|        2663.0|        32760.031|       31547.941|      1100.0|            55.0|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "featuresDf.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get the Latest Version of a Training Dataset (0 if no version exist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "latestVersion: Integer = 2\n",
      "res57: Integer = 2\n"
     ]
    }
   ],
   "source": [
    "val latestVersion = Hops.getLatestTrainingDatasetVersion(\"team_position_prediction\").read()\n",
    "latestVersion"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also explicitly specify optional arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res58: Integer = 2\n"
     ]
    }
   ],
   "source": [
    "Hops.getLatestTrainingDatasetVersion(\"team_position_prediction\").setFeaturestore(Hops.getProjectFeaturestore().read()).read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save as Training Dataset in TFRecords Format"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can create a training dataset from the dataframe with some extended metadata such as schema (automatically inferred). By default when you create a training dataset it will be in \"tfrecords\" format and statistics will be computed for all features. After the dataset have been created you can view and/or update the metadata about the training dataset from the Hopsworks featurestore UI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.createTrainingDataset(\"team_position_prediction\").setDataframe(featuresDf).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also override default parameter values: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.createTrainingDataset(\"team_position_prediction\").setDataframe(featuresDf).setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read()).setVersion(1).setDescription(\"a dataset with features for football teams, used for training a model to predict league-position\").setDataFormat(\"tfrecords\").setDescriptiveStats(false).setFeatureCorr(false).setFeatureHistograms(false).setClusterAnalysis(false).setStatColumns(statColumns).setNumBins(5).setCorrMethod(\"pearson\").setNumClusters(5).write()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "latestVersion: Integer = 1\n",
      "res61: Integer = 1\n"
     ]
    }
   ],
   "source": [
    "val latestVersion = Hops.getLatestTrainingDatasetVersion(\"team_position_prediction_csv\").setFeaturestore(Hops.getProjectFeaturestore.read).read\n",
    "latestVersion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.createTrainingDataset(\"team_position_prediction_csv\").setDataframe(featuresDf).setDataFormat(\"csv\").setDescription(\"a dataset with features for football teams, used for training a model to predict league-position\").write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Create a New Version of A Training Dataset\n",
    "\n",
    "To create a new version, simply use the `createTrainingDataset` method and specify the version argument:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "trainingDatasetVersion: Int = 3\n"
     ]
    }
   ],
   "source": [
    "val trainingDatasetVersion = Hops.getLatestTrainingDatasetVersion(\"team_position_prediction\").read + 1\n",
    "Hops.createTrainingDataset(\"team_position_prediction\").setDataframe(featuresDf).setDataFormat(\"tfrecords\").setDescription(\"a dataset with features for football teams, used for training a model to predict league-position\").setVersion(trainingDatasetVersion).write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inserting Into an Existing Training Dataset\n",
    "\n",
    "Once a dataset have been created, its metadata is browsable in the featurestore registry in the Hopsworks UI. If you don't want to create a new training dataset but just overwrite or insert new data into an existing training dataset, you can use the API function `insertIntoTrainingDataset`.  \n",
    "\n",
    "**Note**: \"append\" write mode is not supported for training datasets stored in tfrecords format, only \"overwrite\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.insertIntoTrainingDataset(\"team_position_prediction_csv\").setDataframe(featuresDf).setMode(\"overwrite\").write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get Training Dataset Path\n",
    "\n",
    "After a **managed dataset** have been created, it is easy to share it and re-use it for training various models. For example if the dataset have been materialized in tf-records format you can call the method `getTrainingDatasetPath()` to get the HDFS path and read it directly in your model training (e.g tensorflow) code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res65: String = hdfs://default/Projects/demo_featurestore_admin000/demo_featurestore_admin000_Training_Datasets/team_position_prediction_csv_1/team_position_prediction_csv\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDatasetPath(\"team_position_prediction_csv\").read"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also override the default arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res66: String = hdfs://default/Projects/demo_featurestore_admin000/demo_featurestore_admin000_Training_Datasets/team_position_prediction_csv_1/team_position_prediction_csv\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDatasetPath(\"team_position_prediction_csv\").setFeaturestore(Hops.getProjectFeaturestore.read).setVersion(Hops.getLatestTrainingDatasetVersion(\"team_position_prediction_csv\").read).read"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read Training Dataset into a Spark Dataframe\n",
    "\n",
    "Typically training datasets are served into deep learning frameworks such as pytorch or tensorflow. However, training datasets can also be read into spark dataframes using the api method `getTrainingDataset()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|team_budget|average_attendance|average_player_age|team_position|sum_attendance|average_player_rating|average_player_worth|sum_player_age|sum_player_rating|sum_player_worth|sum_position|average_position|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|    7307.94|         19595.973|             24.85|            6|     391919.47|            1311.2384|           1435.2465|        2485.0|        131123.84|       143524.64|       563.0|           28.15|\n",
      "|   7326.092|          6462.462|             25.45|           16|     129249.24|             502.2414|           576.87787|        2545.0|         50224.14|       57687.785|       841.0|           42.05|\n",
      "|   3555.235|          7226.672|              25.4|           20|     144533.44|             372.3458|           417.08722|        2540.0|         37234.58|       41708.723|       821.0|           41.05|\n",
      "|  910.39325|         3189.8455|             25.91|           40|      63796.91|            226.29305|            199.3462|        2591.0|        22629.305|       19934.621|      1266.0|            63.3|\n",
      "|  10290.323|         4964.6475|             26.63|           23|     99292.945|             327.6003|            315.4794|        2663.0|        32760.031|       31547.941|      1100.0|            55.0|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDataset(\"team_position_prediction_csv\").read.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also override the default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|team_budget|average_attendance|average_player_age|team_position|sum_attendance|average_player_rating|average_player_worth|sum_player_age|sum_player_rating|sum_player_worth|sum_position|average_position|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "|    7307.94|         19595.973|             24.85|            6|     391919.47|            1311.2384|           1435.2465|        2485.0|        131123.84|       143524.64|       563.0|           28.15|\n",
      "|   7326.092|          6462.462|             25.45|           16|     129249.24|             502.2414|           576.87787|        2545.0|         50224.14|       57687.785|       841.0|           42.05|\n",
      "|   3555.235|          7226.672|              25.4|           20|     144533.44|             372.3458|           417.08722|        2540.0|         37234.58|       41708.723|       821.0|           41.05|\n",
      "|  910.39325|         3189.8455|             25.91|           40|      63796.91|            226.29305|            199.3462|        2591.0|        22629.305|       19934.621|      1266.0|            63.3|\n",
      "|  10290.323|         4964.6475|             26.63|           23|     99292.945|             327.6003|            315.4794|        2663.0|        32760.031|       31547.941|      1100.0|            55.0|\n",
      "+-----------+------------------+------------------+-------------+--------------+---------------------+--------------------+--------------+-----------------+----------------+------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDataset(\"team_position_prediction_csv\").setFeaturestore(Hops.getProjectFeaturestore.read).setVersion(Hops.getLatestTrainingDatasetVersion(\"team_position_prediction_csv\").read).read.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External Training Datasets\n",
    "\n",
    "By default, training datasets in the Feature Store are stored in HopsFS. However, it is also possible to store datasets on S3 and just manage the metadata from Hopsworks Feature Store. \n",
    "\n",
    "To do this, configure an s3 connector for your feature store, e.g with name `my_s3_connector`, then you can save a training dataset to s3 by specifying the s3_connector name in the `sink` argument to `createTrainingDataset`:\n",
    "\n",
    "```scala\n",
    "val storageConnector = \"my_s3_connector\"\n",
    "Hops.createTrainingDataset(training_dataset_name)\n",
    "                          .setDataframe(df)\n",
    "                          .setSink(s3StorageConnectorName).write()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Update Training Dataset Stats\n",
    "\n",
    "The API is similar to the one for updating the stats of a feature group:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateTrainingDatasetStats(\"team_position_prediction\").write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also override the default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateTrainingDatasetStats(\"team_position_prediction\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setVersion(1).setDescriptiveStats(true).setFeatureCorr(true).setFeatureHistograms(true).setClusterAnalysis(true).setStatColumns(null).setNumBins(20).setCorrMethod(\"pearson\").setNumClusters(5).write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get Featurestore Metadata\n",
    "To explore the contents of the featurestore we recommend using the featurestore page in the Hopsworks UI but you can also get the metadata programmatically from the REST API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Update Metadata Cache"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateFeaturestoreMetadataCache().write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also override default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.updateFeaturestoreMetadataCache().setFeaturestore(Hops.getProjectFeaturestore.read).write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### List all Feature Stores Accessible In the Project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res73: java.util.List[String] = [demo_featurestore_admin000_featurestore]\n"
     ]
    }
   ],
   "source": [
    "Hops.getProjectFeaturestores.read"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### List all Feature Groups in a Feature Store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res74: java.util.List[String] = [teams_features_1, games_features_on_demand_tour_1, players_features_1, games_features_hudi_tour_1, games_features_1, attendances_features_1, season_scores_features_1, teams_features_spanish_1, players_features_on_demand_1, teams_features_spanish_3, hudi_featuregroup_test_tour_1, games_features_double_partitioned_1, hudi_featuregroup_test_tour_second_1, hive_fs_sync_example_1, games_features_partitioned_1, hudi_featuregroup_test_tour_1, teams_features_spanish_2, games_features_on_demand_1]\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroups().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Override default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res75: java.util.List[String] = [teams_features_1, games_features_on_demand_tour_1, players_features_1, games_features_hudi_tour_1, games_features_1, attendances_features_1, season_scores_features_1, teams_features_spanish_1, players_features_on_demand_1, teams_features_spanish_3, hudi_featuregroup_test_tour_1, games_features_double_partitioned_1, hudi_featuregroup_test_tour_second_1, hive_fs_sync_example_1, games_features_partitioned_1, hudi_featuregroup_test_tour_1, teams_features_spanish_2, games_features_on_demand_1]\n"
     ]
    }
   ],
   "source": [
    "Hops.getFeaturegroups().setFeaturestore(Hops.getProjectFeaturestore.read).read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### List all Features in a Feature Store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res76: java.util.List[String] = [team_budget, team_id, team_position, average_player_age, average_player_rating, average_player_worth, sum_player_age, sum_player_rating, sum_player_worth, team_id, away_team_id, home_team_id, _hoodie_commit_seqno, _hoodie_commit_time, _hoodie_file_name, _hoodie_partition_path, _hoodie_record_key, score, away_team_id, home_team_id, score, average_attendance, sum_attendance, team_id, average_position, sum_position, team_id, equipo_id, equipo_posicion, equipo_presupuesto, equipo_id, equipo_posicion, equipo_presupuesto, country, id, value, _hoodie_commit_seqno, _hoodie_commit_time, _hoodie_file_name, _hoodie_partition_path, _hoodie_record_key, date, away_team_id, score, home_team_id, country, id, value, _hoodie_commit_seqno, _hoodie_commit_time, _hoodie_file..."
     ]
    }
   ],
   "source": [
    "Hops.getFeaturesList().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Override default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res77: java.util.List[String] = [team_budget, team_id, team_position, average_player_age, average_player_rating, average_player_worth, sum_player_age, sum_player_rating, sum_player_worth, team_id, away_team_id, home_team_id, _hoodie_commit_seqno, _hoodie_commit_time, _hoodie_file_name, _hoodie_partition_path, _hoodie_record_key, score, away_team_id, home_team_id, score, average_attendance, sum_attendance, team_id, average_position, sum_position, team_id, equipo_id, equipo_posicion, equipo_presupuesto, equipo_id, equipo_posicion, equipo_presupuesto, country, id, value, _hoodie_commit_seqno, _hoodie_commit_time, _hoodie_file_name, _hoodie_partition_path, _hoodie_record_key, date, away_team_id, score, home_team_id, country, id, value, _hoodie_commit_seqno, _hoodie_commit_time, _hoodie_file..."
     ]
    }
   ],
   "source": [
    "Hops.getFeaturesList().setFeaturestore(Hops.getProjectFeaturestore.read).read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### List all Training Datasets in a Feature Store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res78: java.util.List[String] = [tour_training_dataset_test_1, team_position_prediction_1, team_position_prediction_csv_1, team_position_prediction_tsv_1, team_position_prediction_parquet_1, team_position_prediction_orc_1, team_position_prediction_avro_1, team_position_prediction_hdf5_1, team_position_prediction_npy_1, team_position_prediction_petastorm_1, team_position_prediction_2, team_position_prediction_3]\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDatasets().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Override default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res79: java.util.List[String] = [tour_training_dataset_test_1, team_position_prediction_1, team_position_prediction_csv_1, team_position_prediction_tsv_1, team_position_prediction_parquet_1, team_position_prediction_orc_1, team_position_prediction_avro_1, team_position_prediction_hdf5_1, team_position_prediction_npy_1, team_position_prediction_petastorm_1, team_position_prediction_2, team_position_prediction_3]\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDatasets().setFeaturestore(Hops.getProjectFeaturestore.read).read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get All Metadata (Features, Feature groups, Training Datasets) for a Feature Store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res80: io.hops.util.featurestore.dtos.app.FeaturestoreMetadataDTO = FeaturestoreMetadataDTO{featurestore=FeaturestoreDTO{featurestoreId=67, featurestoreName='demo_featurestore_admin000_featurestore', created='Wed Oct 30 08:48:07 UTC 2019', hdfsStorePath='hdfs://172.31.22.2:8020/apps/hive/warehouse/demo_featurestore_admin000_featurestore.db', projectName='demo_featurestore_admin000', projectId=119, featurestoreDescription='Featurestore database for project: demo_featurestore_admin000', inodeId=100251}, featuregroups=[CachedFeaturegroupDTO{hiveTableId=6, hdfsStorePaths=[hdfs://172.31.22.2:8020/apps/hive/warehouse/demo_featurestore_admin000_featurestore.db/teams_features_1], inputFormat='org.apache.hadoop.hive.ql.io.orc.OrcInputFormat', hiveTableType=MANAGED_TABLE, inodeId=100932, onlineFe..."
     ]
    }
   ],
   "source": [
    "Hops.getFeaturestoreMetadata().read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Override default parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res81: io.hops.util.featurestore.dtos.app.FeaturestoreMetadataDTO = FeaturestoreMetadataDTO{featurestore=FeaturestoreDTO{featurestoreId=67, featurestoreName='demo_featurestore_admin000_featurestore', created='Wed Oct 30 08:48:07 UTC 2019', hdfsStorePath='hdfs://172.31.22.2:8020/apps/hive/warehouse/demo_featurestore_admin000_featurestore.db', projectName='demo_featurestore_admin000', projectId=119, featurestoreDescription='Featurestore database for project: demo_featurestore_admin000', inodeId=100251}, featuregroups=[CachedFeaturegroupDTO{hiveTableId=6, hdfsStorePaths=[hdfs://172.31.22.2:8020/apps/hive/warehouse/demo_featurestore_admin000_featurestore.db/teams_features_1], inputFormat='org.apache.hadoop.hive.ql.io.orc.OrcInputFormat', hiveTableType=MANAGED_TABLE, inodeId=100932, onlineFe..."
     ]
    }
   ],
   "source": [
    "Hops.getFeaturestoreMetadata().setFeaturestore(Hops.getProjectFeaturestore.read).read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Attach Metadata to a Feature Group"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The feature store enables users to attach a dictionary of metadata to a feature group. It is only supported for CachedFeatureGroups."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.addMetadata(\"teams_features\").setMetadata(Map(\"attr1\" -> \"attr1 value\", \"attr2\" -> \"attr2 value\")).write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get All Metadata associated to a Feature Group"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.getMetadata(\"teams_features\").read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get the metadata by their keys"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.getMetadata(\"teams_features\").setKeys(\"attr1\").read()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Remove metadata attached to a Feature Group"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Hops.removeMetadata(\"teams_features\").setKeys(\"attr1\").write()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## From Raw Data to Features to Training Dataset to Model\n",
    "\n",
    "Once a training dataset have been materialized, we can use it to train a model. In this section we will train an example model using the training dataset `team_position_prediction` that we just created. We will use the column **\"team_position\"** as the target to predict."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Imports\n",
    "\n",
    "In this example we will use Spark MLLib. However, the feature store is in theory agnostic to which framework or method you use for training the model, it works with PyTorch, Tensorflow, MxNet etc."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "import org.apache.spark.ml.feature.VectorAssembler\n",
      "import org.apache.spark.ml.regression.LinearRegression\n"
     ]
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.VectorAssembler\n",
    "import org.apache.spark.ml.regression.LinearRegression"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Constants and HyperParameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NUM_ITER: Int = 1000\n",
      "ELASTIC_REG_PARAM: Double = 0.8\n",
      "REG_LAMBDA_PARAM: Double = 0.3\n"
     ]
    }
   ],
   "source": [
    "val NUM_ITER = 1000\n",
    "val ELASTIC_REG_PARAM = 0.8\n",
    "val REG_LAMBDA_PARAM = 0.3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read TFRecords Dataset into a Spark Dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "dataset_df: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [average_player_rating: float, average_attendance: float ... 10 more fields]\n"
     ]
    }
   ],
   "source": [
    "val dataset_df = Hops.getTrainingDataset(\"team_position_prediction\").read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+------------------+-----------------+------------+----------------+------------------+-------------+--------------------+-----------+----------------+--------------+--------------+\n",
      "|average_player_rating|average_attendance|sum_player_rating|sum_position|sum_player_worth|average_player_age|team_position|average_player_worth|team_budget|average_position|sum_player_age|sum_attendance|\n",
      "+---------------------+------------------+-----------------+------------+----------------+------------------+-------------+--------------------+-----------+----------------+--------------+--------------+\n",
      "|            1311.2384|         19595.973|        131123.84|       563.0|       143524.64|             24.85|            6|           1435.2465|    7307.94|           28.15|        2485.0|     391919.47|\n",
      "|             502.2414|          6462.462|         50224.14|       841.0|       57687.785|             25.45|           16|           576.87787|   7326.092|           42.05|        2545.0|     129249.24|\n",
      "|             372.3458|          7226.672|         37234.58|       821.0|       41708.723|              25.4|           20|           417.08722|   3555.235|           41.05|        2540.0|     144533.44|\n",
      "|            226.29305|         3189.8455|        22629.305|      1266.0|       19934.621|             25.91|           40|            199.3462|  910.39325|            63.3|        2591.0|      63796.91|\n",
      "|             327.6003|         4964.6475|        32760.031|      1100.0|       31547.941|             26.63|           23|            315.4794|  10290.323|            55.0|        2663.0|     99292.945|\n",
      "+---------------------+------------------+-----------------+------------+----------------+------------------+-------------+--------------------+-----------+----------------+--------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also explictly specify the arguments to `getTrainingDataset()` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+------------------+-----------------+------------+----------------+------------------+-------------+--------------------+-----------+----------------+--------------+--------------+\n",
      "|average_player_rating|average_attendance|sum_player_rating|sum_position|sum_player_worth|average_player_age|team_position|average_player_worth|team_budget|average_position|sum_player_age|sum_attendance|\n",
      "+---------------------+------------------+-----------------+------------+----------------+------------------+-------------+--------------------+-----------+----------------+--------------+--------------+\n",
      "|            1311.2384|         19595.973|        131123.84|       563.0|       143524.64|             24.85|            6|           1435.2465|    7307.94|           28.15|        2485.0|     391919.47|\n",
      "|             502.2414|          6462.462|         50224.14|       841.0|       57687.785|             25.45|           16|           576.87787|   7326.092|           42.05|        2545.0|     129249.24|\n",
      "|             372.3458|          7226.672|         37234.58|       821.0|       41708.723|              25.4|           20|           417.08722|   3555.235|           41.05|        2540.0|     144533.44|\n",
      "|            226.29305|         3189.8455|        22629.305|      1266.0|       19934.621|             25.91|           40|            199.3462|  910.39325|            63.3|        2591.0|      63796.91|\n",
      "|             327.6003|         4964.6475|        32760.031|      1100.0|       31547.941|             26.63|           23|            315.4794|  10290.323|            55.0|        2663.0|     99292.945|\n",
      "+---------------------+------------------+-----------------+------------+----------------+------------------+-------------+--------------------+-----------+----------------+--------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Hops.getTrainingDataset(\"team_position_prediction\").setSpark(spark).setFeaturestore(Hops.getProjectFeaturestore.read).setVersion(1).read().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convert the Dataframe into Spark MLLib data format\n",
    "\n",
    "Spark MLLib models typically expect all features to be grouped into a single column instead of having one column per feature, we can use Spark's `VectorAssembler` to group our features together"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- average_player_rating: float (nullable = true)\n",
      " |-- average_attendance: float (nullable = true)\n",
      " |-- sum_player_rating: float (nullable = true)\n",
      " |-- sum_position: float (nullable = true)\n",
      " |-- sum_player_worth: float (nullable = true)\n",
      " |-- average_player_age: float (nullable = true)\n",
      " |-- team_position: long (nullable = true)\n",
      " |-- average_player_worth: float (nullable = true)\n",
      " |-- team_budget: float (nullable = true)\n",
      " |-- average_position: float (nullable = true)\n",
      " |-- sum_player_age: float (nullable = true)\n",
      " |-- sum_attendance: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset_df.printSchema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "transformedDf: org.apache.spark.sql.DataFrame = [team_position: bigint, features: vector]\n"
     ]
    }
   ],
   "source": [
    "val transformedDf = new VectorAssembler().\n",
    "  setInputCols(Array( \"average_player_rating\",\"average_attendance\", \"sum_player_rating\", \n",
    "                     \"sum_position\", \"sum_player_worth\", \"average_player_age\", \"average_player_worth\",\n",
    "                    \"team_budget\", \"average_position\", \"sum_player_age\", \"sum_attendance\")).\n",
    "  setOutputCol(\"features\").\n",
    "  transform(dataset_df).\n",
    "    drop(\"average_player_rating\").\n",
    "    drop(\"average_attendance\").\n",
    "    drop(\"sum_player_rating\").\n",
    "    drop(\"sum_player_worth\").\n",
    "    drop(\"average_player_age\").\n",
    "    drop(\"average_player_worth\").\n",
    "    drop(\"team_budget\").\n",
    "    drop(\"average_position\").\n",
    "    drop(\"sum_player_age\").\n",
    "    drop(\"sum_attendance\").\n",
    "    drop(\"sum_position\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- team_position: long (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "transformedDf.printSchema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define The Model Using Spark MLLib\n",
    "\n",
    "We will use a linear regression model. In this tutorial we work with so little data that using a larger model does not make sense."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lr: org.apache.spark.ml.regression.LinearRegression = linReg_43dc07226d5c\n"
     ]
    }
   ],
   "source": [
    "val lr = new LinearRegression().\n",
    "    setLabelCol(\"team_position\").\n",
    "    setFeaturesCol(\"features\").\n",
    "    setMaxIter(NUM_ITER).\n",
    "    setRegParam(REG_LAMBDA_PARAM).\n",
    "    setElasticNetParam(ELASTIC_REG_PARAM)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Train The Model using The Parsed Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lrModel: org.apache.spark.ml.regression.LinearRegressionModel = linReg_43dc07226d5c\n"
     ]
    }
   ],
   "source": [
    "val lrModel = lr.fit(transformedDf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Show Model Training Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------+------------------+\n",
      "|            features|team_position|        prediction|\n",
      "+--------------------+-------------+------------------+\n",
      "|[1311.23840332031...|            6| 5.466254654943459|\n",
      "|[502.241394042968...|           16|18.323688419895568|\n",
      "|[372.345794677734...|           20|17.566396134916936|\n",
      "|[226.293045043945...|           40|37.123184809082446|\n",
      "|[327.600311279296...|           23|29.786095913268838|\n",
      "|[205.155975341796...|           39|35.657752460940586|\n",
      "|[881.298278808593...|            9|11.262826067715793|\n",
      "|[467.793792724609...|           17| 16.82646856958825|\n",
      "|[657.393188476562...|           11| 9.628936445599663|\n",
      "|[231.505447387695...|           33|31.792601813546227|\n",
      "|[322.697967529296...|           26|29.921433049367856|\n",
      "|[297.791961669921...|           27| 31.64619079360343|\n",
      "|[562.398986816406...|           15|10.368459419413938|\n",
      "|[181.494277954101...|           43|  46.6363093515945|\n",
      "|[229.579666137695...|           36| 36.37811651821866|\n",
      "|[298.033020019531...|           24|21.299690263521413|\n",
      "|[389.657623291015...|           21|  22.5713111745358|\n",
      "|[7191.86328125,92...|            1|1.0090139372799953|\n",
      "|[178.769195556640...|           46|46.202186808737174|\n",
      "|[150.963272094726...|           48|48.322429007238675|\n",
      "+--------------------+-------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "// print the output column and the input column and the truth label\n",
    "lrModel.transform(transformedDf).select(\"features\", \"team_position\", \"prediction\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "trainingSummary: org.apache.spark.ml.regression.LinearRegressionTrainingSummary = org.apache.spark.ml.regression.LinearRegressionTrainingSummary@6822a7b9\n",
      "numIterations: 56\n",
      "objectiveHistory: [0.5,0.41124102893970266,0.1251899443343994,0.08777706436856089,0.03913658284734845,0.039023660048732915,0.03902163129558637,0.03901820891749766,0.03899877146854527,0.03899662259028954,0.03898657050161849,0.038985197140997234,0.03898517236997781,0.03898515439811365,0.038985123255113,0.03898510959220626,0.03898506916169116,0.03898504112882173,0.03898500402608166,0.038984965224528455,0.038984887713267934,0.03898469837571618,0.03898315598281196,0.03898302661971978,0.038981810415314816,0.038981644705092855,0.038980633751504776,0.038980055488513785,0.03897885683972229,0.038977553371827484,0.038977353664081,0.03897705829014849,0.03897700336935213,0.03897696008492962,0.038976909759634984,0.038976888388403345,0.038976872666554675,0.038976818625222745,0.038976773319026714,0.038976764382013085,0.03897676123679094,0.03897675951478595,0.038976758242250935,0.038976757866287404,0.03897675768455181,0.038976757617913714,0.03897675759655468,0.03897675758512237,0.038976757582232735,0.03897675758024323,0.03897675757972829,0.038976757579505326,0.03897675757940496,0.03897675757933291,0.03897675757931109,0.03897675757930404]\n",
      "+--------------------+\n",
      "|           residuals|\n",
      "+--------------------+\n",
      "|  0.5337453450565413|\n",
      "|  -2.323688419895568|\n",
      "|  2.4336038650830645|\n",
      "|  2.8768151909175543|\n",
      "|  -6.786095913268838|\n",
      "|  3.3422475390594144|\n",
      "| -2.2628260677157925|\n",
      "| 0.17353143041174945|\n",
      "|  1.3710635544003367|\n",
      "|  1.2073981864537728|\n",
      "|  -3.921433049367856|\n",
      "|  -4.646190793603431|\n",
      "|   4.631540580586062|\n",
      "|  -3.636309351594498|\n",
      "| -0.3781165182186612|\n",
      "|  2.7003097364785873|\n",
      "| -1.5713111745358006|\n",
      "|-0.00901393727999...|\n",
      "|-0.20218680873717432|\n",
      "|-0.32242900723867507|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "RMSE: 3.002426880708375\n",
      "r2: 0.9567127626698669\n"
     ]
    }
   ],
   "source": [
    "val trainingSummary = lrModel.summary\n",
    "println(s\"numIterations: ${trainingSummary.totalIterations}\")\n",
    "println(s\"objectiveHistory: [${trainingSummary.objectiveHistory.mkString(\",\")}]\")\n",
    "trainingSummary.residuals.show()\n",
    "println(s\"RMSE: ${trainingSummary.rootMeanSquaredError}\")\n",
    "println(s\"r2: ${trainingSummary.r2}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark",
   "language": "",
   "name": "sparkkernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}